package com.demo.virtual_thread.selfRealize;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

//调度器（负责调度载体的执行）
public class ResumableScheduler {

	private final Thread schedulerThread;
	private final Queue<ResumableCarrier> runnableQueue = new ConcurrentLinkedQueue<>();
	private final Queue<ResumableCarrier> suspendedQueue = new ConcurrentLinkedQueue<>();
	private final AtomicBoolean isRunning = new AtomicBoolean(false);
	// 跟踪所有提交的载体，包括正在执行的
    private final Set<ResumableCarrier> allCarriers = ConcurrentHashMap.newKeySet();
    
	public ResumableScheduler(String name) {
		this.schedulerThread = new Thread(this::schedule, name);
	}

	public void start() {
		isRunning.set(true);
		schedulerThread.start();
	}

	public void stop() {
		isRunning.set(false);
		schedulerThread.interrupt();
	}

	public void submit(ResumableCarrier carrier) {
		runnableQueue.offer(carrier);
		allCarriers.add(carrier); // 加入全局跟踪
	}

	private void schedule() {
		System.out.println("调度器启动:" + Thread.currentThread().getName());

		while (isRunning.get()) {
			// 检查挂起队列，恢复可执行的载体
			checkAndResumeSuspended();

			// 执行可运行队列中的载体
			ResumableCarrier carrier = runnableQueue.poll();
			if (carrier != null) {
				try {
					System.out.println("调度器开始执行载体:" + carrier.getName());
					carrier.execute(); // 从当前步骤继续执行
				} catch (SuspendException e) {
					// 挂起后放入挂起队列
					suspendedQueue.offer(carrier);
				}
			} else {
				// 无任务时短暂休眠
				try {
					TimeUnit.MILLISECONDS.sleep(10);
				} catch (InterruptedException e) {
					Thread.currentThread().interrupt();
					break;
				}
			}
		}

		System.out.println("调度器停止:" + Thread.currentThread().getName());
	}

	// 检查并恢复挂起的载体（模拟挂起结束）
	private void checkAndResumeSuspended() {
		if (!suspendedQueue.isEmpty()) {
			// 每次恢复 1 个挂起的载体（实际场景可根据条件判断）
			ResumableCarrier carrier = suspendedQueue.poll();
			if (carrier != null) {
				carrier.resume();
				runnableQueue.offer(carrier);
				System.out.println("载体" + carrier.getName() + "恢复执行");
			}
		}
	}

	// 判断所有任务是否完成
	public boolean isAllCompleted() {
		return allCarriers.stream().allMatch(ResumableCarrier::isCompleted);
	}
}